package com.example.testkafkaproto.consumer.interceptor;

import org.apache.kafka.clients.consumer.ConsumerInterceptor;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.util.Map;
import java.util.Set;

public class TestConsumerInterceptor implements ConsumerInterceptor<String,String> {

    @Override
    public ConsumerRecords<String, String> onConsume(ConsumerRecords<String, String> consumerRecords) {

        Set<TopicPartition> tps = consumerRecords.partitions();
        for(TopicPartition tp : tps){
            System.out.println("=====拦截消费:topic:" + tp.topic() + ",partition:" + tp.partition() + ",有" + consumerRecords.records(tp).size() + "条消息");
        }
        return consumerRecords;
    }

    @Override
    public void onCommit(Map<TopicPartition, OffsetAndMetadata> map) {

        for(TopicPartition tp : map.keySet()){
            OffsetAndMetadata data = map.get(tp);
            System.out.println("=====拦截提交:topic:" + tp.topic() + ",partition:" + tp.partition() + ",提交位移为" + data.offset());
        }
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> map) {

    }

}
